package com.glodon.paas.framework.registry.server;

import com.glodon.paas.framework.registry.core.type.ClientBasic;
import com.glodon.paas.framework.registry.core.type.ClientDetail;
import com.glodon.paas.framework.registry.core.type.ServiceBasic;
import com.glodon.paas.framework.registry.core.type.ServiceDetail;
import com.glodon.paas.framework.registry.core.zk.SessionStateProcessor;
import com.glodon.paas.framework.registry.core.zk.SessionStateWatcher;
import com.glodon.paas.framework.registry.core.zk.ZooKeeperUtils;
import com.glodon.paas.framework.registry.server.health.NodeScanner;
import org.apache.commons.lang.SerializationUtils;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * @author Don Li
 */
public class ServerManager {

    private static final Logger LOGGER = LoggerFactory.getLogger(ServerManager.class);
    private static final int DEFAULT_RETRIES = 3;

    private String connectString;
    private int sessionTimeout;
    private int retries = DEFAULT_RETRIES;
    private String servicesPath;
    private NodeScanner nodeScanner;

    private ZooKeeper zk;

    public ServerManager(String connectString, int sessionTimeout) {
        this.connectString = connectString;
        this.sessionTimeout = sessionTimeout;
    }

    public void init() {
        List<Watcher.Event.KeeperState> states = new ArrayList<Watcher.Event.KeeperState>();
        states.add(Watcher.Event.KeeperState.SyncConnected);
        states.add(Watcher.Event.KeeperState.Expired);
        final SessionStateWatcher watcher = new SessionStateWatcher();
        watcher.setRegisteredState(states);
        watcher.setProcessor(new SessionStateProcessor() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {
                    LOGGER.info("ServerManager connected to zk cluster");
                } else if (event.getState() == Watcher.Event.KeeperState.Expired) {
                    LOGGER.info("ServerManager zk client session expired, now create a new zk client");
                    zk = ZooKeeperUtils.createZooKeeper(connectString, sessionTimeout, watcher, retries);
                    nodeScanner.scanAll();
                }
            }
        });

        zk = ZooKeeperUtils.createZooKeeper(connectString, sessionTimeout, watcher, retries);
        nodeScanner.setManager(this);
        nodeScanner.scanAll();
    }

    public void setRetries(int retries) {
        this.retries = retries;
    }

    public ZooKeeper getZk() {
        return zk;
    }

    public void setNodeScanner(NodeScanner nodeScanner) {
        this.nodeScanner = nodeScanner;
    }

    public void setServicesPath(String servicesPath) {
        this.servicesPath = servicesPath;
    }

    public List<String> listAllServices() {
        try {
            return zk.getChildren(servicesPath, false);
        } catch (Exception e) {
            LOGGER.warn("Fail to list all services");
        }
        return null;
    }

    public ServiceDetail getServiceDetail(String serviceId) {
        String path = servicesPath + "/" + serviceId;
        Stat serviceStat = new Stat();
        try {
            ServiceBasic basic = (ServiceBasic) SerializationUtils.deserialize(zk.getData(path, false, serviceStat));
            ServiceDetail detail = new ServiceDetail(basic, serviceStat);
            // get client details and build the service detail
            List<ClientDetail> clientDetails = new ArrayList<ClientDetail>();
            if (serviceStat.getNumChildren() > 0) {
                List<String> clients = zk.getChildren(path, false);
                for (String client : clients) {
                    Stat clientStat = new Stat();
                    ClientBasic clientBasic = (ClientBasic)
                            SerializationUtils.deserialize(zk.getData(path + "/" + client, false, clientStat));
                    ClientDetail clientDetail = new ClientDetail(clientBasic, clientStat);
                    clientDetails.add(clientDetail);
                }
            }
            detail.setClients(clientDetails);
            return detail;
        } catch (Exception e) {
            LOGGER.error("Error getting data for znode", e);
        }
        return null;
    }
}
